package com.aotain.project.apollo.bolt;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Base64;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.IPDatabase;
import com.aotain.apollo.ServerInfo;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.storm.AbstractBolt;
import com.aotain.storm.TupleHelpers;

/**
 * 端口服务扫描
 * @author Administrator
 *
 */
public class PortServiceScanBolt extends AbstractBolt{

//	HashMap<Integer,PortInfo> ports = null;
	
	
	
	HTable _hTable = null; 
			
	
	HashMap<String,ServerInfo> _mapServers = null; 
	
	HashMap<Long,IPDatabase> map = null;
	
	HashMap<String,Integer> _mapDIPandPort 
		= new HashMap<String,Integer>();
	
	HashMap<String,Integer> _mapPortScanStat 
		= new HashMap<String,Integer>();
	
	String _curDate = "";
	
	/**
	 * 常用服务端口
	 */
	HashMap<String,HashMap<Integer,Integer>> _mapServicePort = 
			new HashMap<String,HashMap<Integer,Integer>>();
	
	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			if(TupleHelpers.isTickTuple(tuple)) {
//				emitCountingData();  
				
				/**加载前一天的端口服务配置数据,记录常用服务端口，其他端口为异常端口
				记录某个端口在五分钟内被访问的次数，设定阀值，如果不是前一天常用访问端口，提出告警
				//function**/
				SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
				String curDate = sdf_d.format(new Date());
				if(!_curDate.equals(curDate))
				{//跨天加载配置
					LoadServicePort();
					_curDate = curDate;
				}
				PortScanDetect();
				collector.ack(tuple);
			} else {
				//
				PortScanCount(tuple);
				//PortScanDetect(tuple);
//				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
//			System.err.println(" PortDetectBolt exception"+e.getMessage());
			Logger.getLogger("STDERR").error("PortServiceScanBolt exception====",e);
			collector.fail(tuple);
			throw new FailedException("PortServiceScanBolt exception", e);
		}	
	}
	
	private void PortScanDetect()
	{
		try {
			
			
			for(Entry<String,Integer> entry : _mapPortScanStat.entrySet())
			{
				String key = entry.getKey();
				String values[] = key.split("\\|");
				String destip = values[0];
				String sourceip = values[1];
				String destport = values[2];
				String gisstr = values[3];
				String desc = values[4];
				String score = values[5];
				String idc = values[6];
				
				String[] gisArray = gisstr.split("#");
				String destAreaName = gisArray[0]; //如果地市为空，精确到国家
				String DestGeo = gisArray[1];
				String sourceAreaName = gisArray[2];  //如果地市为空，精确到国家
				String SourceGeo = gisArray[3];
				String sourceAreaCountry = gisArray[4];
				String sourceProvinceID =  gisArray[5];
				String sourceProvinceName =  gisArray[6]; //如果省为空，精确到国家
				String sourceAreaCityId = gisArray[7];
				String sourceAreaProvinceId = gisArray[8];
				
				HBaseRecordAdd addDestIP = HBaseRecordAdd.getInstance(zooserver);
				
				

				Date current = new Date();
				SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmmss");
				String strDate = sdf.format(current);
				
				String rowkey = String.format("%s_%s_%s_%s_%d", destip, strDate, 
						sourceip,destport,3);
				
				String tbName = ApolloProperties.SDS_ABNORMAL_LOG;
				String cf = "cf";
				
		    	addDestIP.Add(tbName, rowkey, cf,  "DESTIP", destip);
		    	addDestIP.Add(tbName, rowkey, cf,  "DESTPORT", String.valueOf(destport));
		    	addDestIP.Add(tbName, rowkey, cf,  "SOURCEIP", sourceip);
		    	addDestIP.Add(tbName, rowkey, cf,  "ACCESSTIME", strDate);
		    	//addDestIP.Add(tbName, rowkey, cf,  "DOMAIN", domainname);
		    	//addDestIP.Add(rowkey, "SOURCEPORT", sourceport);
		    	addDestIP.Add(tbName, rowkey, cf, "ABRNORMAL", "4");
		    	addDestIP.Add(tbName, rowkey, cf,  "EVALUATE", String.valueOf(score));
		    	
//		    	if(destport.equals("22") &&_hTable != null )
//				{
//					Get get = new Get(destip.getBytes());  
//			        Result rs = _hTable.get(get);  
//			        for(Cell cc : rs.getColumnCells("cf".getBytes(), "PORT22".getBytes())){  
//			        	desc = "恶意端口扫描[" + destport + "]";
//			        }  
//				}
		    	
		    	addDestIP.Add(tbName, rowkey, cf,  "DESC", desc);
		    	
		    	addDestIP.Add(tbName, rowkey, cf,  "ATTNUM", String.valueOf(entry.getValue()));
		    	
		    	addDestIP.Add(tbName, rowkey, cf,  "SOURCEAREA", sourceAreaName.replace("市", ""));
		    	addDestIP.Add(tbName, rowkey, cf,  "SOURCEGEO", String.format("%s",SourceGeo));
		    	
		    	addDestIP.Add(tbName, rowkey, cf,  "DESTAREA", String.format("%s",destAreaName.replace("市", "")));
		    	addDestIP.Add(tbName, rowkey, cf,  "DESTGEO", String.format("%s",DestGeo));
		    	
		    	addDestIP.Add(tbName, rowkey, cf, "SOURCECOUNTRY", sourceAreaCountry);
		    	
		    	
		    	
		    	//##########COPY LOG
		    	String tbName2 = ApolloProperties.SDS_ABNORMAL_LOG2;
		    	addDestIP.Add(tbName2, rowkey, cf,  "DESTIP", destip);
		    	addDestIP.Add(tbName2, rowkey, cf,  "DESTPORT", String.valueOf(destport));
		    	addDestIP.Add(tbName2, rowkey, cf,  "SOURCEIP", sourceip);
		    	addDestIP.Add(tbName2, rowkey, cf,  "ACCESSTIME", strDate);
		    	addDestIP.Add(tbName2, rowkey, cf, "ABRNORMAL", "4");
		    	addDestIP.Add(tbName2, rowkey, cf,  "EVALUATE", String.valueOf(score));
		    	addDestIP.Add(tbName2, rowkey, cf,  "DESC", desc);
		    	addDestIP.Add(tbName2, rowkey, cf,  "ATTNUM", String.valueOf(entry.getValue()));
		    	addDestIP.Add(tbName2, rowkey, cf,  "SOURCEAREA", sourceAreaName.replace("市", ""));
		    	addDestIP.Add(tbName2, rowkey, cf,  "SOURCEGEO", String.format("%s",SourceGeo));
		    	addDestIP.Add(tbName2, rowkey, cf,  "DESTAREA", String.format("%s",destAreaName.replace("市", "")));
		    	addDestIP.Add(tbName2, rowkey, cf,  "DESTGEO", String.format("%s",DestGeo));
		    	addDestIP.Add(tbName2, rowkey, cf, "SOURCECOUNTRY", sourceAreaCountry);
		    	
		    	String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
		    	String rowkey3 = String.format("%s_%s", strDate, sourceip);
		    	addDestIP.Add(tbName3, rowkey3, "cf", "ACCESSTIME", strDate);
		    	addDestIP.Add(tbName3, rowkey3, "cf", "SOURCEIP", sourceip);
		    	addDestIP.Add(tbName3, rowkey3, "cf", "SOURCEGEO", String.format("%s",SourceGeo));
		    	addDestIP.Add(tbName3, rowkey3, "cf", "ATTNUM", String.valueOf(entry.getValue()));
		    	addDestIP.Add(tbName3, rowkey3, "cf", "AREAID", sourceProvinceID);
		    	addDestIP.Add(tbName3, rowkey3, "cf", "AREANAME", sourceProvinceName);
		    	addDestIP.Add(tbName3, rowkey3, "cf", "CITYID", sourceAreaCityId);
		    	addDestIP.Add(tbName3, rowkey3, "cf", "CITYNAME", sourceAreaName);
		    	addDestIP.Add(tbName3, rowkey3, "cf", "DESTIP", destip);
		    	addDestIP.Add(tbName3, rowkey3, "cf", "DESTPORT", String.valueOf(destport));
		    	addDestIP.Add(tbName3, rowkey3, "cf", "TYPE", "4");
		    	addDestIP.Add(tbName3, rowkey3, "cf", "SOURCECOUNTRY", sourceAreaCountry);
		    	addDestIP.Add(tbName3, rowkey3, cf,  "DESTGEO", String.format("%s",DestGeo));
		    	addDestIP.Add(tbName3, rowkey3, "cf", "DESTAREA", destAreaName);
		    	
		    	SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
				SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
				SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
		    	String dateStr_h = sdf_h.format(current);
				String dateStr_d = sdf_d.format(current);
				String dateStr_hour = sdf_hour.format(current);
				
		    	collector.emit(new Values(destip,sourceip,String.valueOf(destport), "",
		    			dateStr_d,dateStr_h,sourceProvinceID,sourceAreaName,
		    			sourceProvinceName, entry.getValue(), dateStr_hour,"PORTLOW", idc));
		    	
			}
			
			_mapPortScanStat.clear();
			} catch (Exception e) {
				// TODO Auto-generated catch block
//				System.err.println(" PortDetectBolt exception"+e.getMessage());
				Logger.getLogger("STDERR").error("PortServiceScanBolt exception====",e);
				throw new FailedException(e);
			}
	}
	
	/**
	 * 判断端口是否是非常用端口
	 * @param tuple
	 */
	private void PortScanCount(Tuple tuple)
	{
		try
		{
			String destip = tuple.getString(0);
			int destport = Integer.parseInt(tuple.getString(1));
			boolean isNormal = true;
			
			if(_mapServicePort.containsKey(destip))
			{
				HashMap<Integer,Integer> portMap = _mapServicePort.get(destip);
				if(!portMap.containsKey(destport))
				{
					isNormal = false;  //非常用端口
					Logger.getLogger(PortDetectBolt.class).info("##### Port Is Noraml ###["+destport+"]");
				}
			}
			
			if(isNormal == true)
			{
				collector.ack(tuple);
				return;
			}
			
			
			String sourceip = tuple.getString(2);
			byte[] gisbytes =  Base64.decodeBase64(tuple.getString(4));
			String gisstr = new String(gisbytes);
			String idc = tuple.getStringByField("idcno");


		
			String desc = "";
			int score = 100;
			if(!isNormal)
			{
				desc = "非常用端口被连接["+destport+"].";
				score = 60;
			}
			
			String key = String.format("%s|%s|%d|%s|%s|%d|%s",	
					destip, sourceip, destport, gisstr, desc, score, idc);
			
			int count = 1;
			if(_mapPortScanStat.containsKey(key))
				count = _mapPortScanStat.get(key) + 1;
			_mapPortScanStat.put(key, count);
		} catch (Exception e) {
			// TODO Auto-generated catch block
//			System.err.println("PortDetectBolt emitCountingData exception"+e.getMessage());
			Logger.getLogger("STDERR").error("PortServiceScanBolt PortScanCount exception====",e);
			throw new FailedException("PortServiceScanBolt PortScanCount exception====", e);
		}
		
	}
	
	/**
	 * 加载服务端口配置信息
	 */
	private void LoadServicePort()
	{
		Configuration hConfig = HBaseConfiguration.create();
        hConfig.set("hbase.zookeeper.quorum", zooserver);
        hConfig.set("hbase.zookeeper.property.clientPort","2181"); 
        Connection connection;
        try {
        	connection = ConnectionFactory.createConnection(hConfig);
        	TableName TABLE_NAME = TableName.valueOf("SDS_SERVICE_IP_PORT");
        	_hTable = (HTable) connection.getTable(TABLE_NAME);
        	
        	_mapServicePort.clear();
        	SimpleDateFormat dfDay = new SimpleDateFormat("yyyyMMdd");
			Date dStartTimeDay = new Date();
			String strDateDay = dfDay.format(dStartTimeDay);
			
			Calendar c = Calendar.getInstance();
			c.add(Calendar.DATE, -1);// 1 minute ago
			String strEndDay = dfDay.format(c.getTime());
			
			Logger.getLogger(PortDetectBolt.class).info(String.format("###Scan Port [%s] [%s]",strEndDay,strDateDay));
			
        	Scan scan = new Scan(strEndDay.getBytes(),strDateDay.getBytes());
        	
        	
        	ResultScanner rs = _hTable.getScanner(scan);
        	HashMap<String,HashMap<Integer,Long>> map 
        	 	 = new HashMap<String,HashMap<Integer,Long>>();
            //for (Result r : rs) {
        	Result r = null;
            while ((r = rs.next()) != null) {
            	String IP = "";
        		Integer Port = 0;
        		Long count = 0L;
            	for(Cell cell : r.listCells())
            	{
            		//String ff = Bytes.toString(CellUtil.cloneFamily(cell));
	  				String cc = Bytes.toString(CellUtil.cloneQualifier(cell));
	  				 
            		if(cc.equals("IP"))
            		{
            			IP = Bytes.toString(CellUtil.cloneValue(cell));
            		}
            		else if(cc.equals("PORT"))
            		{
            			String sPort = Bytes.toString(CellUtil.cloneValue(cell));
            			if(sPort.trim().isEmpty())
            				continue;
            			Port = Integer.parseInt(sPort);
            		}
            		else if(cc.equals("SESSCOUNT"))
            		{
            			count = Bytes.toLong(CellUtil.cloneValue(cell));
            		}
            	}
            	
            	if(map.containsKey(IP))
            	{
            		HashMap<Integer,Long> portMap = map.get(IP);
            		portMap.put(Port,count);
            	}
            	else
            	{
            		HashMap<Integer,Long> portMap = new HashMap<Integer,Long>();
            		portMap.put(Port,count);
            		map.put(IP, portMap);
            	}
            }
            
            rs.close();
            _hTable.close();
            
            //对每一个IP的服务端口排序
            for(Entry<String,HashMap<Integer,Long>> entry : map.entrySet())
            {
            	String IP = entry.getKey();
            	HashMap<Integer,Long> ports = entry.getValue();
            	Long total = 0L;
            	for(Entry<Integer,Long> e : ports.entrySet())
            	{
            		total = total + e.getValue();
            	}
            	
            	for(Entry<Integer,Long> e : ports.entrySet())
            	{
//            		String info = String.format("###Normal Port### %s:[%s] %f", IP, e.getKey(), (double)e.getValue()/(double)total);
//            		Logger.getRootLogger().info(info);
            		if((double)e.getValue()/(double)total >= 0.1) //超过30% 认为是正常服务端口
            		{
            			if(_mapServicePort.containsKey(IP))
            			{
            				HashMap<Integer,Integer> p1 = _mapServicePort.get(IP);
            				p1.put(e.getKey(), e.getKey());
            			}
            			else
            			{
            				HashMap<Integer,Integer> 
            					p1 = new HashMap<Integer,Integer>();
            				p1.put(e.getKey(), e.getKey());
            				
            				_mapServicePort.put(IP,p1);
            			}
//            			Logger.getRootLogger().info("###Normal Port###" + IP 
//            					+ ":" + e.getKey());
            		}
            	}
            }
        
        } catch (Exception e) {
				// TODO Auto-generated catch block
//				e.printStackTrace();
			Logger.getLogger("STDERR").error("PortServiceScanBolt ERROR:", e);
			throw new FailedException("PortServiceScanBolt ERROR:", e);
		}
	}



	@Override
	public void declareOutputFields(OutputFieldsDeclarer arg0) {
		// TODO Auto-generated method stub
		arg0.declare(new Fields("dip","sip","dport","sport","dateStr_d",
				"dateStr_h","sourceAreaId","sourceAreaName","areaName",
				"pv","dateStr_hour","abntype", "idcno"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,600);
		return conf;
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		_curDate = sdf_d.format(new Date());
		LoadServicePort(); //初始化加载服务IP配置
		
		Configuration hConfig = HBaseConfiguration.create();
        hConfig.set("hbase.zookeeper.quorum", zooserver);
        hConfig.set("hbase.zookeeper.property.clientPort","2181"); 
		 
        Connection connection;
		try {//加载黑名单数据信息
			connection = ConnectionFactory.createConnection(hConfig);
			 TableName TABLE_NAME = TableName.valueOf("SDS_IP_BLACKLIST");
			 _hTable = (HTable) connection.getTable(TABLE_NAME);
				
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}
